package org.util.concurrent;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author liangsu
 * @version v1.0
 * @Description
 * @Date 2020/11/6 9:16
 * @since
 */
public class BindThreadExecutor implements Executor {
//    private final Logger logger = LoggerFactory.getLogger(BindThreadExecutor.class);

    private static final RejectedExecutionHandler defaultHandler =
            new ThreadPoolExecutor.AbortPolicy();

    /** 核心线程数 */
    private int coreSize;

    /** 工作线程的队列创建工厂 */
    private BindThreadQueueFactory queueFactory;
    /** 线程工厂 */
    private ThreadFactory threadFactory;

    private RejectedExecutionHandler handler;

    /** 工作线程 */
    private ThreadPoolExecutor[] executors;
    /** 工作线程的队列 */
    private BlockingQueue<Runnable>[] blockingQueues;

    private AtomicInteger threadSelector = new AtomicInteger(0);

    public BindThreadExecutor(int coreSize){
        this(coreSize, Executors.defaultThreadFactory(),
                defaultHandler,
                BindThreadQueueFactory.DEFAULT);
    }

    /**
     *
     * @param coreSize
     * @param threadFactory
     * @param handler
     * @param queueFactory
     */
    public BindThreadExecutor(int coreSize,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler,
                              BindThreadQueueFactory queueFactory){
        this.coreSize = coreSize;
        this.executors = new ThreadPoolExecutor[coreSize];
        this.blockingQueues = new BlockingQueue[coreSize];
        this.threadFactory = threadFactory;
        this.handler = handler;
        this.queueFactory = queueFactory;
    }

    /**
     * 提交任务
     * @param command
     */
    @Override
    public void execute(Runnable command) {
        execute(command, threadSelector.getAndIncrement());
    }

    /**
     * 将任务绑定到指定的线程上执行
     * @param command 任务
     * @param threadId 线程id，用于决定在那个线程上执行任务
     */
    public void execute(Runnable command, int threadId) {
        if (command == null)
            throw new NullPointerException();

        if(threadId < 0){
            throw new IllegalArgumentException("thread id exception: " + threadId);
        }
        // 计算在哪个线程上执行
        int index = index(threadId);

        // 创建线程
        if(executors[index] == null){
            synchronized (executors){
                if(executors[index] == null){
                    // 创建队列
                    blockingQueues[index] = queueFactory.newQueue();

                    // 单线程的线程池
                    ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1,
                            0L, TimeUnit.MILLISECONDS,
                            blockingQueues[index],
                            threadFactory,
                            handler);
                    executors[index] = executor;
                }
            }
        }

        // 提交任务
        executors[index].execute(command);
    }

    protected int index(int threadId){
        return threadId % executors.length;
    }

}
